package com.atguigu.pro

import java.net.URL

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable

/**
 *
 * @description: 统计页面uv
 * @time: 2021-03-30 21:48
 * @author: baojinlong
 * */



object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 方便测试全局并行度为1
    env.setParallelism(1)

    val resource: URL = getClass.getResource("/UserBehavior-short.csv")
    // 从文件中读取数据
    // val inputStream: DataStream[String] = env.readTextFile(resource.getPath)
    // C:/codes/scala/FlinkTutorial/src/main/resources/UserBehavior.csv  E:/big-data/FlinkTutorial/src/main/resources/UserBehavior.csv
    val inputStream: DataStream[String] = env.readTextFile("C:/codes/scala/FlinkTutorial/src/main/resources/UserBehavior.csv")

    // 转换成样例类类型并提取时间戳和watermark
    val dataStream: DataStream[UserBehavior2] =
      inputStream
        .map(data => {
          val dataArray: Array[String] = data.split(",")
          UserBehavior2(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
        })
        .assignAscendingTimestamps(_.timestamp * 1000)

    val uvCountResult: DataStream[UvCount] = dataStream
      .filter(x => "pv".equals(x.behavior))
      .timeWindowAll(Time.hours(1)) // 直接不分组,基于DataStream开1小时滚动窗口,并行度为1可以直接全窗口统计,如果并行度不是1的话就需要根据windowEnd分组处理
      .apply(new VuCountResult)
    uvCountResult.print("uvCountResult")
    env.execute("pvJobTest")

  }
}

// 定义输出UV统计样例类
case class UvCount(windowEnd: Long, count: Long)

// 自定义实现全窗口函数,用一个set结构来保存所有数据,进行自动去重,注意全窗口函数会把所有数据都保存成状态数据
class VuCountResult extends AllWindowFunction[UserBehavior2, UvCount, TimeWindow] {
  override def apply(window: TimeWindow, input: Iterable[UserBehavior2], out: Collector[UvCount]): Unit = {
    // 定义一个Set
    val userIdSet: mutable.Set[Long] = mutable.Set[Long]()
    // 遍历窗口中所有数据,把userId添加到set中,自动去重
    for (elem <- input) {
      userIdSet += elem.userId
    }
    // 将set的size作为去重后的uv值输出
    out.collect(UvCount(window.getEnd, userIdSet.size))
  }
}